use cross_rs::{memory::bytes_io::BytesIO, string::extend::StringExentd, IResult};
use redis_rs::RedisPool;

use crate::zip::xzip;


pub fn hash_dump(dir_dest: &str, redis_addr: String, db_num_keys_str: &str) -> IResult<String> {
    if db_num_keys_str.len() < 1 {
        error!("redis_mod::hash_dump：没有指定db_num_keys_str.");
        return Ok("redis_mod::hash_dump：没有指定db_num_keys_str.".to_string());
    }
    let mut keys_指定 = db_num_keys_str.split_arr(',');
    let db_num = keys_指定[0].parse::<u8>()?;
    keys_指定.remove(0);

    let file_dest = format!("{dir_dest}/{db_num:0>2}.redis.xz");
    warn!("Start, redis_mod::hash_dump: to {file_dest}, from redis db_num={db_num}, keys_指定={keys_指定:?}, {redis_addr}");
    let redis_pool = RedisPool::new(redis_addr.clone(), vec![db_num])?;
    let mut buf = BytesIO::new("hash_dump".to_string());

    let keys = redis_pool.keys(db_num, "*")?;
    if keys.len() == 0 {
        return Ok(file_dest);
    }
    warn!("\t1. {} keys to dump...", keys.len());
    let mut count_key = 0;
    let mut count_field = 0;
    for key in keys.iter() {
        count_key += 1;
        if keys_指定.len() > 0 && !keys_指定.contains(&key) {
            continue;
        }
        let map = redis_pool.hget_all_bytes(db_num, key)?;
        if map.len() < 1 {
            continue;
        }
        count_field += map.len();
        if count_key % 50 == 0 {
            warn!("\t\t{count_key} keys, {count_field} fields");
        }

        buf.write_utf8_with_size(key);
        buf.write_i32_le(map.len() as i32);
        for (field, value) in map.iter() {
            buf.write_utf8_with_size(field);
            buf.write_blob_with_size(&value);
        }
    }
    warn!("\t2. deal: {count_key} keys, {count_field} fields");

    warn!("\t3. xzipping..., {file_dest}");
    xzip::zip_bytesio_to_file(buf, &file_dest, 6)?;

    warn!("End, redis_mod::hash_dump: to {file_dest}, from redis db_num={db_num} {redis_addr}");

    Ok(file_dest)
}


pub fn hash_restore(redis_addr: String, db_num: u8, dir_src: &str) -> IResult {
    let file_src = format!("{dir_src}/{db_num:0>2}.redis.xz");
    warn!("Start, redis_mod::hash_restore: from {file_src}, to redis db_num={db_num} {redis_addr}");
    let redis_pool = RedisPool::new(redis_addr.clone(), vec![db_num])?;

    let bts = xzip::unzip_file_to_bytes(&file_src)?;
    let mut buf = BytesIO::from_bytes("hash_restore".to_string(), bts);

    let mut count_key = 0;
    let mut count_field = 0;
    loop {
        let key = match buf.read_utf8_with_size() {
            Ok(v) => v.1,
            Err(e) => {
                error!("\tfinised: may be end of file. {e:?}");
                break;
            },
        };
        let len_field = buf.read_i32_le()?;
        count_key += 1;
        count_field += len_field;
        
        if count_key % 50 == 0 {
            warn!("\t\t{count_key} keys, {count_field} fields");
        }

        let mut list_pairs = vec![];
        for _ in 0..len_field {
            let (_sz, field) = buf.read_utf8_with_size()?;
            let (_sz, value) = buf.read_utf8_with_size()?;
            list_pairs.push((field, value));
        }

        redis_pool.hmset_pairs(db_num, key, list_pairs)?;
    }
    warn!("\t2. deal total: {count_key} keys, {count_field} fields");

    warn!("End, redis_mod::hash_restore: from {file_src}, to redis db_num={db_num} {redis_addr}");

    Ok(())
}

